/*
 *  (c) E.S.A.
 */
#include <pthread.h>
#include <iostream>
#include <cassert>
#include <vector>
using namespace std;

#include "process_data.h"
#include "msg_logger.h"
#include "tokenizer.h" 
#include "remote_party.h"
#include "global_data.h"
#include "client_mode.h"


global_s global;

vector<pthread_t> thread_vect;

/*
 * data format --> 
 * 		from <ip_address>\n to <ip_address>\n body <body_of_the_message>
 * 
 */
void process_data (const std::string& data)
{
	String_Tokenizer data_tok;
	std::string from_ip, to_ip, tmp_str;

	MSG_LOG (DBG, "message processed " << data);

	data_tok.set_string(data, " \t\n");

	if (data_tok.has_more_tokens() == false) {
		MSG_LOG (WARN, "Empty message");
		return;
	}

	//assert(data_tok.has_more_tokens());

	tmp_str = data_tok.next_token();
	if (tmp_str == "from:"){
		// data receive from remote host
		from_ip = data_tok.next_token();
		data_tok.next_token(); // skip "to:"
		to_ip = data_tok.next_token();
		// to_ip should be his own ip
		if (to_ip != global.local_addr.ip){
			MSG_LOG (WARN, "TO value should be the same with the local address");
			return;
		}
		// a client with ip = "127.0.0.1" must be connected to socket_engine
		r_client_data *remote_client = global.list_of_clients.get_element("127.0.0.1");
		if (remote_client == NULL)
			MSG_LOG (WARN, "Warning: The manager is not yet connected to the socket_engine");
		// @p: add data in cli_send_buff
		sem_wait( &global.cli_send_buff.empty_sem );
		sem_wait( &global.cli_send_buff.put_mutex );

		global.cli_send_buff.mess[global.cli_send_buff.nextin] = data;
		global.cli_send_buff.nextin = (global.cli_send_buff.nextin + 1) % CIRCULAR_BUF_SIZE;

		sem_post( &global.cli_send_buff.put_mutex );
		sem_post( &global.cli_send_buff.full_sem );

	} else

		if (tmp_str == "to:"){
			// data receive from localhost

			// add from field to the message
			std::string processed_data("from: ");
			processed_data.append(global.local_addr.ip);
			processed_data.append("\n");
			processed_data.append(data);

			// to whom the message must be send
			to_ip = data_tok.next_token();

			// is any client with this ip already connected to S_E ?
			r_client_data *remote_client = global.list_of_clients.get_element(to_ip);
			if (remote_client != NULL) {
				// @p: add data in cli_send_buff
				sem_wait( &global.cli_send_buff.empty_sem );
				sem_wait( &global.cli_send_buff.put_mutex );

				global.cli_send_buff.mess[global.cli_send_buff.nextin] = processed_data;
				global.cli_send_buff.nextin = (global.cli_send_buff.nextin + 1) % CIRCULAR_BUF_SIZE;

				sem_post( &global.cli_send_buff.put_mutex );
				sem_post( &global.cli_send_buff.full_sem );
			}
			else { // S_E already connected to the server with this ip?
				remote_srv *remote_server = global.list_of_servers.get_element(to_ip);
				if (remote_server == NULL){
					MSG_LOG (WARN, "Warning: Node unknown!");
					return;
				}
				if (remote_server->get_conn_state() == CONN){
					remote_server->add_data_to_buffer(processed_data);
				} 
				else { // not -> create a new client_thread for the corresponding server ip
					pthread_t client_thread;
					remote_server->add_data_to_buffer(processed_data);
					if(pthread_create(&client_thread, NULL, client_thread_main, remote_server) != 0){
						MSG_LOG (ERR, "could not create server conn thread");
						return;
					}
					thread_vect.push_back(client_thread);
				}
			}

		}

}

void* process_data_main (void* arg)
{
	std::string tmp_str;

	MSG_LOG (DBG, "process_data thread starting ..");

	// @c
	while (!global.quit_flag) {
		if ( sem_trywait( &global.recv_buff.full_sem ) == 0 ) { 
			sem_wait( &global.recv_buff.get_mutex);

			MSG_LOG (DBG,"process_data");

			tmp_str = global.recv_buff.mess[global.recv_buff.nextout];
			process_data(tmp_str);

			global.recv_buff.nextout = (global.recv_buff.nextout + 1) % CIRCULAR_BUF_SIZE;

			sem_post ( &global.recv_buff.get_mutex);
			sem_post ( &global.recv_buff.empty_sem);

		}

	}

	// wait for joing all client threads:
	typedef vector<pthread_t>::iterator v_iterator;
	for(v_iterator it = thread_vect.begin(); it != thread_vect.end(); ++it){
		if(pthread_join(*it, NULL) != 0){
			MSG_LOG (ERR, "could not join client thread");
			return NULL;
		}
	}

	MSG_LOG (INFO, "process_data thread finishing .. ");
	return NULL;
}


